package com.hub.realtime.resource.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.hub.realtime.common.core.domain.model.ClusterInfo;
import com.hub.realtime.common.utils.StringUtils;
import com.hub.realtime.common.utils.hadoop.HdfsUtil;
import com.hub.realtime.framework.service.ServiceBase;
import com.hub.realtime.resource.domain.ResFlinkJob;
import com.hub.realtime.resource.domain.ResSavepoint;
import com.hub.realtime.resource.dto.input.SavePointParam;
import com.hub.realtime.resource.mapper.ResSavepointMapper;
import com.hub.realtime.resource.service.IResApplicationService;
import com.hub.realtime.resource.service.IResFlinkJobService;
import com.hub.realtime.resource.service.IResSavepointService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;

@Service
@Slf4j
public class ResSavepointServiceImpl extends ServiceBase<ResSavepointMapper, ResSavepoint> implements IResSavepointService {

    @Autowired
    IResFlinkJobService flinkJobService;

    @Autowired
    IResApplicationService applicationService;


    @Override
    public Page<ResSavepoint> list(SavePointParam param) {
        LambdaQueryWrapper<ResSavepoint> wrapper = new LambdaQueryWrapper<>();
        wrapper.orderByDesc(ResSavepoint::getCreateTime);
        if (StringUtils.isNotEmpty(param.getJobName())) {
            wrapper.like(ResSavepoint::getJobName, param.getJobName());
        }
        return getPageList(param, wrapper);
    }

    @Override
    public int deleteInfos(Long[] ids) {
        try {
            for (long id : ids) {
                ResSavepoint resSavepoint = baseMapper.selectById(id);
                Long jobId = resSavepoint.getJobId();
                ResFlinkJob resFlinkJob = flinkJobService.getBaseMapper().selectById(jobId);
                Long applicationId = resFlinkJob.getApplicationId();
                ClusterInfo clusterInfo = applicationService.getClusterInfoByAppId(applicationId);
                HdfsUtil.delete(clusterInfo, resSavepoint.getPath());
            }
        } catch (Exception ex) {

        }
        return delByIds(ids);
    }
}
